import pandas as pd
import numpy as np
from sqlalchemy import create_engine
import pymysql
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Optional, Dict, Any
import logging
from logging.handlers import TimedRotatingFileHandler
from datetime import datetime, timedelta
import os
from contextlib import asynccontextmanager

# 配置日志
LOG_DIR = "logs"
os.makedirs(LOG_DIR, exist_ok=True)

log_file = os.path.join(LOG_DIR, "recommend.log")
handler = TimedRotatingFileHandler(
    log_file, when="midnight", interval=1, backupCount=7, encoding="utf-8"
)
formatter = logging.Formatter(
    "%(asctime)s - %(levelname)s - %(message)s", datefmt="%Y-%m-%d %H:%M:%S"
)
handler.setFormatter(formatter)

logger = logging.getLogger("recommend_logger")
logger.setLevel(logging.INFO)
logger.addHandler(handler)

console_handler = logging.StreamHandler()
console_handler.setFormatter(formatter)
logger.addHandler(console_handler)

DB_CONFIG = {
    'host': 'localhost',
    'port': 3306,
    'user': 'root',
    'password': '123456',
    'database': 'mydb'
}

# Pydantic模型定义
class RecommendationRequest(BaseModel):
    user_id: str
    top_n: Optional[int] = 20

class SimpleRecommendationResponse(BaseModel):
    code: int
    result: List[str]
    msg: str

class DataLoader:
    def __init__(self):
        self.video_df = None
        self.user_df = None
        self.viewlogs_df = None  # 用户性别数据从viewlogs表获取
        self.ks = [0.8, 1.2, 0.2, 1.5, 1, 1.3, 1.4, 1.1, 0.6, 0.7]
        self.engine = None
        self.user_id_field = "user_id"  # 默认用户ID字段名
        
    def create_connection(self):
        """创建数据库连接"""
        try:
            self.engine = create_engine(
                f"mysql+pymysql://{DB_CONFIG['user']}:{DB_CONFIG['password']}@{DB_CONFIG['host']}:{DB_CONFIG['port']}/{DB_CONFIG['database']}",
                pool_pre_ping=True
            )
            logger.info("数据库连接成功")
            return True
        except Exception as e:
            logger.error(f"数据库连接失败: {str(e)}")
            return False
            
    def load_viewlogs_data(self):
        """从viewlogs表加载用户性别数据"""
        try:
            # 从viewlogs表获取用户性别信息
            viewlogs_query = "SELECT DISTINCT user_id, user_sex FROM viewlogs WHERE user_sex IS NOT NULL"
            self.viewlogs_df = pd.read_sql(viewlogs_query, self.engine)
            logger.info(f'成功从viewlogs表加载用户性别数据: {len(self.viewlogs_df)}条')
            # 记录viewlogs表的字段信息
            logger.info(f"viewlogs表字段: {self.viewlogs_df.columns.tolist()}")
            return True
        except Exception as e:
            logger.warning(f"从viewlogs表加载用户性别数据失败: {str(e)}")
            self.viewlogs_df = pd.DataFrame()
            return False

    def get_user_watched_videos(self, user_id):
        """获取用户已观看的视频列表"""
        try:
            # 根据实际表结构调整查询
            watched_query = f"""
            SELECT DISTINCT video_id 
            FROM user_watch_history 
            WHERE user_id = '{user_id}'
            """
            watched_df = pd.read_sql(watched_query, self.engine)
            watched_videos = watched_df['video_id'].astype(str).tolist()
            logger.info(f"用户 {user_id} 已观看视频数量: {len(watched_videos)}")
            return watched_videos
        except Exception as e:
            logger.warning(f"获取用户 {user_id} 观看记录失败: {str(e)}，返回空列表")
            return []

    def calculate_time_decay(self, publish_time, base_date=None):
        """计算时间衰减因子"""
        if base_date is None:
            base_date = datetime.now()
        
        if isinstance(publish_time, str):
            try:
                publish_time = datetime.strptime(publish_time, '%Y-%m-%d %H:%M:%S')
            except:
                publish_time = datetime.strptime(publish_time, '%Y-%m-%d')
        
        days_diff = (base_date - publish_time).days
        decay = np.exp(-days_diff / 30)  # 30天衰减系数
        return max(0.1, min(1.0, decay))  # 限制在0.1-1.0之间

    def load_video_with_labels(self):
        """加载视频和标签数据"""
        try:
            # 1.1 读取 video 表（视频基础信息）
            video_query = "SELECT * FROM video"
            video_df = pd.read_sql(video_query, self.engine)
            if video_df.empty:
                logger.error("video 表中无数据，无法加载视频信息")
                return None
            logger.info(f"从 video 表加载基础视频数据: {len(video_df)}条")

            # 1.2 读取 video_label 表（视频标签数据）
            label_query = "SELECT * FROM video_label"
            label_df = pd.read_sql(label_query, self.engine)
            if not label_df.empty:
                logger.info(f"从 video_label 表加载标签数据: {len(label_df)}条")
                # 适配 video_label 表字段名
                label_df.columns = label_df.columns.str.strip()
                if "video id" in label_df.columns:
                    label_df.rename(columns={"video id": "video_id"}, inplace=True)
                if "label_ name" in label_df.columns:
                    label_df.rename(columns={"label_ name": "label_name"}, inplace=True)
                if "video_id" in label_df.columns and "label_name" in label_df.columns:
                    # 按 video_id 聚合标签（去重）
                    label_agg = label_df.groupby("video_id")["label_name"].agg(
                        lambda x: list(set([str(tag).strip() for tag in x if pd.notna(tag)]))
                    ).reset_index()
                    label_agg.columns = ["video_id", "video_tags_list"]
                    # 左连接视频表和标签表
                    video_df = pd.merge(video_df, label_agg, on="video_id", how="left")
                else:
                    logger.warning("video_label 表缺少必要字段，无法处理标签")
            else:
                logger.warning("video_label 表中无数据，所有视频初始无标签")
                video_df["video_tags_list"] = [[] for _ in range(len(video_df))]  # 空标签列表兜底

            # 1.3 填充无标签的视频，统一字段类型
            video_df["video_tags_list"] = video_df["video_tags_list"].apply(
                lambda x: x if isinstance(x, list) else []
            )
            # 1.4 统一视频ID字段名：保持为video_id
            logger.info(f"视频数据字段: {video_df.columns.tolist()}")

            return video_df
        except Exception as e:
            logger.error(f"加载视频和标签数据失败: {str(e)}")
            return None

    def load_data(self):
        if not self.create_connection():
            return False
            
        try:
            # -------------------------- 1. 加载视频+标签数据--------------------------
            # 执行视频+标签加载
            self.video_df = self.load_video_with_labels()
            if self.video_df is None or self.video_df.empty:
                logger.error("无法加载视频数据（含标签），服务无法提供推荐")
                return False
            logger.info(f"成功加载视频数据（含标签）: {len(self.video_df)}条，其中有标签视频: {self.video_df[self.video_df['video_tags_list'].str.len() > 0].shape[0]}条")

            # -------------------------- 2. 加载用户画像数据--------------------------
            user_query = "SELECT * FROM user_portraits"
            self.user_df = pd.read_sql(user_query, self.engine)
            logger.info(f'成功从数据库加载用户画像数据: {len(self.user_df)}条')
            
            # 记录实际字段名，自动匹配用户ID字段
            logger.info(f"用户画像表字段名: {self.user_df.columns.tolist()}")
            if self.user_id_field not in self.user_df.columns:
                possible_fields = ["user_id", "uid", "userid", "u_id", "id", "userID", "UserID"]
                found = False
                for field in possible_fields:
                    if field in self.user_df.columns:
                        self.user_id_field = field
                        found = True
                        logger.info(f"用户画像表中使用'{field}'作为用户ID字段")
                        break
                if not found:
                    logger.error(f"用户画像表中未找到用户ID相关字段，可用字段: {self.user_df.columns.tolist()}")
                    return False

            # 将用户ID转换为字符串（适配图2的text类型）
            self.user_df[self.user_id_field] = self.user_df[self.user_id_field].astype(str).str.strip()

            # -------------------------- 3. 从viewlogs表加载用户性别数据--------------------------
            if not self.load_viewlogs_data():
                logger.warning("无法从viewlogs表加载用户性别数据，性别相关的标签权重调整功能将不可用")
            
            return True
        except Exception as e:
            logger.error(f"从数据库加载数据失败: {str(e)}")
            return False

# 初始化数据加载器
data_loader = DataLoader()

# 使用新的lifespan事件处理器替代过时的on_event
@asynccontextmanager
async def lifespan(app: FastAPI):
    # 启动时加载数据
    logger.info("开始加载数据...")
    if data_loader.load_data():
        logger.info("数据加载完成，服务准备就绪")
    else:
        logger.error("数据加载失败，服务可能无法正常工作")
    yield
    # 关闭时可以添加清理逻辑
    logger.info("服务关闭")

# 创建FastAPI应用，使用新的lifespan处理器
app = FastAPI(
    title="视频推荐API服务", 
    description="基于用户标签偏好的视频推荐系统，排除已观看视频",
    lifespan=lifespan
)

# ------------------------------ 推荐核心逻辑 ------------------------------
def get_user_recommendations(user_id, top_n=20, temperature=1.5, gender_boost_factor=1.5):
    """
    获取用户推荐视频的核心逻辑
    """
    logger.info(f"开始为用户 {user_id} 生成推荐，top_n={top_n}")
    
    # 1. 获取用户已观看视频
    watched_videos = data_loader.get_user_watched_videos(user_id)
    
    # 2. 从viewlogs表获取用户性别信息
    user_sex = "unknown"
    if data_loader.viewlogs_df is not None and not data_loader.viewlogs_df.empty:
        try:
            # 适配viewlogs表的字段名
            viewlogs_id_field = "user_id"
            if "user_id" not in data_loader.viewlogs_df.columns:
                possible_fields = ["user_id", "uid", "userid", "u_id", "id"]
                for field in possible_fields:
                    if field in data_loader.viewlogs_df.columns:
                        viewlogs_id_field = field
                        break
            
            user_sex_data = data_loader.viewlogs_df[
                data_loader.viewlogs_df[viewlogs_id_field].astype(str) == str(user_id)
            ]
            if not user_sex_data.empty and "user_sex" in user_sex_data.columns:
                sex_value = user_sex_data.iloc[0]["user_sex"]
                # 处理性别值：1表示男，0表示女
                if sex_value == 1:
                    user_sex = "male"
                elif sex_value == 0:
                    user_sex = "female"
                logger.info(f"用户 {user_id} 性别: {user_sex} (user_sex={sex_value})")
        except Exception as e:
            logger.warning(f"从viewlogs表获取用户 {user_id} 性别失败: {str(e)}")
    
    # 3. 获取用户画像数据
    user_portrait = None
    if data_loader.user_df is not None and not data_loader.user_df.empty:
        user_portrait = data_loader.user_df[
            data_loader.user_df[data_loader.user_id_field].astype(str) == str(user_id)
        ]
    
    # 4. 如果用户不存在于画像数据中，返回热门视频
    if user_portrait is None or user_portrait.empty:
        logger.info(f"用户 {user_id} 不存在于画像数据中，返回热门视频")
        return get_hot_videos(watched_videos, top_n)
    
    # 5. 解析用户偏好标签
    user_prefer_tags = {}
    adjusted_tags_info = {}
    
    try:
        user_data = user_portrait.iloc[0]
        
        # 5.1 处理labels字段（用户偏好标签）
        if "labels" in user_data and pd.notna(user_data["labels"]):
            labels_str = str(user_data["labels"]).strip()
            if labels_str:
                try:
                    # 尝试解析JSON格式的标签
                    import json
                    labels_dict = json.loads(labels_str)
                    if isinstance(labels_dict, dict):
                        user_prefer_tags = labels_dict
                except:
                    # 如果不是JSON，尝试其他格式解析
                    labels_list = [tag.strip() for tag in labels_str.split(',') if tag.strip()]
                    for tag in labels_list:
                        user_prefer_tags[tag] = user_prefer_tags.get(tag, 0) + 1
        
        # 5.2 处理category字段（用户偏好分类）
        if "category" in user_data and pd.notna(user_data["category"]):
            category_str = str(user_data["category"]).strip()
            if category_str:
                categories = [cat.strip() for cat in category_str.split(',') if cat.strip()]
                for cat in categories:
                    user_prefer_tags[cat] = user_prefer_tags.get(cat, 0) + 1
        
        # 5.3 根据用户行为数据调整标签权重
        behavior_weights = {
            "likes": 1.2,
            "favorites": 1.5, 
            "reposts": 1.3,
            "completions": 1.1,
            "play_rate": 1.0
        }
        
        for behavior, weight in behavior_weights.items():
            if behavior in user_data and pd.notna(user_data[behavior]):
                try:
                    behavior_value = float(user_data[behavior])
                    # 对用户已有的所有标签按行为数据加权
                    for tag in user_prefer_tags:
                        user_prefer_tags[tag] += behavior_value * weight * 0.01
                except:
                    pass
        
        # 5.4 根据用户性别调整标签权重
        if user_sex in ["male", "female"]:
            gender_tags = {
                "male": ["游戏", "科技", "体育", "汽车", "军事"],
                "female": ["美妆", "时尚", "母婴", "美食", "情感"]
            }
            for tag in gender_tags.get(user_sex, []):
                if tag in user_prefer_tags:
                    user_prefer_tags[tag] *= gender_boost_factor
                    adjusted_tags_info[tag] = f"性别调整 x{gender_boost_factor}"
        
        # 5.5 应用温度参数增加多样性
        if user_prefer_tags and temperature != 1.0:
            max_score = max(user_prefer_tags.values())
            for tag in user_prefer_tags:
                # 应用softmax温度调节
                user_prefer_tags[tag] = (user_prefer_tags[tag] / max_score) ** (1/temperature) * max_score
        
        logger.info(f"用户 {user_id} 解析出的偏好标签数量: {len(user_prefer_tags)}")
        
    except Exception as e:
        logger.error(f"解析用户 {user_id} 偏好标签时出错: {str(e)}")
        return get_hot_videos(watched_videos, top_n)
    
    # 6. 匹配视频标签与用户偏好
    dfv = data_loader.video_df.copy()
    
    # 6.1 排除已观看视频
    if watched_videos:
        # 确保视频ID类型一致
        dfv["video_id"] = dfv["video_id"].astype(str).str.strip()
        dfv = dfv[~dfv["video_id"].isin(watched_videos)]
        if len(dfv) == 0:
            logger.warning(f"用户 {user_id} 已观看所有视频，返回空推荐结果")
            return {
                "user_prefer_tags": user_prefer_tags,
                "sorted_prefer_tags": sorted(user_prefer_tags.items(), key=lambda x: x[1], reverse=True)[:10],
                "selected_videos": pd.DataFrame(),
                "user_sex": user_sex,
                "adjusted_tags": adjusted_tags_info,
                "watched_count": len(watched_videos)
            }
    
    # 6.2 补充标签：若 video_tags_list 为空，用 category 字段兜底
    if 'category' in dfv.columns:
        dfv["video_tags_list"] = dfv.apply(
            lambda row: [str(row["category"]).strip()] if (not row["video_tags_list"] and pd.notna(row["category"])) else row["video_tags_list"],
            axis=1
        )
        logger.info(f"用户 {user_id}：用 category 补充空标签视频后，有标签视频数量: {dfv[dfv['video_tags_list'].str.len() > 0].shape[0]}条")
    else:
        logger.warning("video 表缺少 category 字段，无法为无标签视频补充分类标签")
    
    # 6.3 计算视频匹配得分
    def calculate_video_score(row):
        # 标签匹配得分：用户偏好标签与视频标签的权重总和
        tag_score = 0
        if isinstance(row.get("video_tags_list"), list):
            for tag in row["video_tags_list"]:
                tag_score += user_prefer_tags.get(str(tag).strip(), 0)
        
        # 时间衰减因子
        if 'publish_time' in row and pd.notna(row["publish_time"]):
            time_decay = data_loader.calculate_time_decay(row["publish_time"])
        else:
            time_decay = 1.0  # 无发布时间时不衰减
        
        return tag_score * time_decay
    
    dfv["match_score"] = dfv.apply(calculate_video_score, axis=1)
    
    # 6.4 按匹配得分降序排序，取前top_n条
    dfv_sorted = dfv.sort_values(by="match_score", ascending=False).reset_index(drop=True)
    actual_top_n = min(top_n, len(dfv_sorted))
    dfv_selected = dfv_sorted.head(actual_top_n)
    
    # 6.5 整理用户偏好标签（取前10个权重最高的标签）
    sorted_prefer_tags = sorted(user_prefer_tags.items(), key=lambda x: x[1], reverse=True)[:10]
    
    logger.info(f"用户 {user_id} 推荐完成，返回 {len(dfv_selected)} 个视频")
    
    return {
        "user_prefer_tags": user_prefer_tags,
        "sorted_prefer_tags": sorted_prefer_tags,
        "selected_videos": dfv_selected,
        "user_sex": user_sex,
        "adjusted_tags": adjusted_tags_info,
        "watched_count": len(watched_videos)
    }

def get_hot_videos(watched_videos, top_n=20):
    """返回热门视频（按likes+favourites+reposts排序）"""
    try:
        dfv = data_loader.video_df.copy()
        
        # 排除已观看视频
        if watched_videos:
            dfv["video_id"] = dfv["video_id"].astype(str).str.strip()
            dfv = dfv[~dfv["video_id"].isin(watched_videos)]
        
        # 计算热度得分（如果相关字段存在）
        if all(col in dfv.columns for col in ['likes', 'favorites', 'reposts']):
            dfv['hot_score'] = dfv['likes'].fillna(0) + dfv['favorites'].fillna(0) * 1.5 + dfv['reposts'].fillna(0) * 1.3
            dfv_sorted = dfv.sort_values(by='hot_score', ascending=False)
        else:
            # 如果没有行为数据，随机选择
            dfv_sorted = dfv.sample(frac=1)
        
        actual_top_n = min(top_n, len(dfv_sorted))
        dfv_selected = dfv_sorted.head(actual_top_n)
        
        return {
            "user_prefer_tags": {},
            "sorted_prefer_tags": [],
            "selected_videos": dfv_selected,
            "user_sex": "unknown",
            "adjusted_tags": {},
            "watched_count": len(watched_videos)
        }
    except Exception as e:
        logger.error(f"获取热门视频失败: {str(e)}")
        return {
            "user_prefer_tags": {},
            "sorted_prefer_tags": [],
            "selected_videos": pd.DataFrame(),
            "user_sex": "unknown",
            "adjusted_tags": {},
            "watched_count": len(watched_videos)
        }

# ------------------------------ FastAPI 接口定义 ------------------------------
@app.get("/")
async def root():
    return {"message": "视频推荐API服务（支持个性化推荐/热门视频推荐）", "status": "running"}

@app.post("/recommend", response_model=SimpleRecommendationResponse)
async def recommend_videos(request: RecommendationRequest):
    """
    单用户推荐接口
    - 若用户存在：返回基于标签偏好的个性化推荐（实时排除已观看）
    - 若用户不存在：返回按 likes+favourites+reposts 排序的热门视频（实时排除已观看）
    - 固定返回20个视频ID，temperature固定为1.5
    """
    try:
        # 调用核心推荐逻辑，固定参数：top_n=20, temperature=1.5, gender_boost_factor=1.5
        result = get_user_recommendations(
            user_id=request.user_id,
            top_n=request.top_n or 20,  # 使用请求的top_n或默认20
            temperature=1.5,    # 固定多样性参数为1.5
            gender_boost_factor=1.5  # 固定性别加强系数为1.5
        )
        
        # 提取视频ID列表
        video_ids = []
        if not result["selected_videos"].empty:
            video_ids = [str(row["video_id"]) for _, row in result["selected_videos"].iterrows()]
        
        return SimpleRecommendationResponse(
            code=200,
            result=video_ids,
            msg=f"success (已排除{result.get('watched_count', 0)}个已观看视频)"
        )
    except Exception as e:
        logger.error(f"为用户 {request.user_id} 生成推荐时出错: {str(e)}")
        return SimpleRecommendationResponse(
            code=500,
            result=[],
            msg=f"推荐服务异常: {str(e)}"
        )

@app.get("/recommend/all")
async def recommend_all_users(top_n: int = 20, gender_boost_factor: float = 1.5):
    """批量为所有用户生成推荐（用户不存在时自动跳过）"""
    try:
        # 获取所有有效用户ID
        if data_loader.user_df is None or data_loader.user_id_field not in data_loader.user_df.columns:
            return {
                "code": 500,
                "result": {},
                "msg": "用户画像数据未正确加载，无法进行批量推荐"
            }
            
        user_ids = data_loader.user_df[data_loader.user_id_field].unique()
        results = {}
        processed_count = 0
        
        for user_id in user_ids:
            try:
                # 为单个用户生成推荐，使用固定temperature=1.5
                user_result = get_user_recommendations(
                    user_id=user_id, 
                    top_n=top_n, 
                    temperature=1.5,  # 批量推荐也使用固定temperature
                    gender_boost_factor=gender_boost_factor
                )
                # 提取视频ID列表
                video_ids = []
                if not user_result["selected_videos"].empty:
                    video_ids = [str(row["video_id"]) for _, row in user_result["selected_videos"].iterrows()]
                
                results[str(user_id)] = {
                    "code": 200,
                    "result": video_ids,
                    "msg": f"success (已排除{user_result.get('watched_count', 0)}个已观看视频)"
                }
                processed_count += 1
            except Exception as e:
                logger.error(f"处理用户 {user_id} 时出错: {str(e)}，已跳过")
                results[str(user_id)] = {
                    "code": 500,
                    "result": [],
                    "msg": f"处理用户时出错: {str(e)}"
                }
                continue
        
        return {
            "processed_users": processed_count,
            "results": results,
            "timestamp": datetime.now().isoformat()
        }
    except Exception as e:
        logger.error(f"批量推荐时出错: {str(e)}")
        return {
            "code": 500,
            "result": {},
            "msg": f"批量推荐服务异常: {str(e)}"
        }

if __name__ == "__main__":
    import uvicorn
    # 启动服务（默认端口8102，允许外部访问）
    uvicorn.run(app, host="0.0.0.0", port=8102, log_level="info")